package pack01_udtf;

import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDTF;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;

import java.util.ArrayList;
import java.util.List;
/*
   "zookeeper,hadoop,hdfs,hive,MapReduce"    -----> zookeeper
                                                    hadoop
                                                    hdfs
                                                    hive
                                                    MapReduce
 */
public class MyUDTF extends GenericUDTF {
    private final  Object[] forwardListObj = new Object[1];

    /*
       1、该方法只会执行一次，用来做初始化
       2、该方法用来指定你转换完之后每一列名字和每一列的类型
     */
    @Override
    public StructObjectInspector initialize(StructObjectInspector argOIs) throws UDFArgumentException {
        //设置列名的类型
        List<String> fieldNames = new ArrayList();

        //设置列名，就是转换完之后列名字
        fieldNames.add("column_01");
        //fieldNames.add("column_02");

        //设置列的类型
        List<ObjectInspector> fieldOIs = new ArrayList<ObjectInspector>()  ;

        //设置输出的列的值类型
        fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
        //fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);

        return ObjectInspectorFactory.getStandardStructObjectInspector(fieldNames, fieldOIs);

    }


    /*
       1、表中的数据有多少行需要处理，该方法就会执行多少次
       2、函数最后如何调用
        select my_udtf("zookeeper,hadoop,hdfs,hive,MapReduce",",") word;
            objects[0] ---->"zookeeper,hadoop,hdfs,hive,MapReduce"
            objects[1] ----> ","
     */
    @Override
    public void process(Object[] objects) throws HiveException {
        //1:获取原始数据
        String args = objects[0].toString(); // "zookeeper,hadoop,hdfs,hive,MapReduce"

        //2:获取数据传入的第二个参数，此处为分隔符
        String splitKey = objects[1].toString();  // ","


        //3.将原始数据按照传入的分隔符进行切分
        String[] fields = args.split(splitKey);  // "zookeeper,hadoop,hdfs,hive,MapReduce".split(',')


        //4:遍历切分后的结果，并写出
        for (String word : fields) {  // [zookeeper,hadoop,hdfs,hive,MapReduce]
            //将每一个单词添加值对象数组
            forwardListObj[0] = word;
            //forwardListObj[1] = word;
            //将对象数组内容写出
            forward(forwardListObj);
        }

    }

    @Override
    public void close() throws HiveException {

    }
}
